package com.xxk.rocket.star.core.producer;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.MQProducer;
import com.xxk.rocket.star.core.context.ProducerContext;
import com.xxk.rocket.star.core.exception.StargateException;
import com.xxk.rocket.star.core.utils.ProcessID;
import com.xxk.rocket.star.core.utils.StargateEnvironment;
import com.xxk.rocket.star.core.utils.StringUtils;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class ProducerFactory {

  private static final Map<String, MQProducer> producerMap = new ConcurrentHashMap<String, MQProducer>();
  private static final Object lock = new Object();

  public static MQProducer getProducer(ProducerContext producerContext)
          throws MQClientException {
    return getProducer(producerContext.getGroup(), producerContext.getNamesrvAddr(), false);
  }

  public static MQProducer getProducerAndStart(String group, String namesrvAddr)
          throws MQClientException {
    return getProducer(group, namesrvAddr, true);
  }

  public static MQProducer getProducer(String group, String namesrvAddr, boolean start)
          throws MQClientException {
    if (StringUtils.isBlank(group) || StringUtils.isBlank(namesrvAddr) ||
            StargateEnvironment.getNamesrv(namesrvAddr) == null) {
      throw new StargateException("nameServer or group is null");
    }
    group = group + "-" + namesrvAddr;
    synchronized (lock) {
      MQProducer mqProducer = producerMap.get(group);
      if (mqProducer != null) {
        return mqProducer;
      }
      DefaultMQProducer producer = new DefaultMQProducer(group);
      producer.setNamesrvAddr(StargateEnvironment.getNamesrv(namesrvAddr));
      producer.setInstanceName(ProcessID.getPID() + "-" + namesrvAddr);
      producerMap.put(group, producer);
      if (start) {
        producer.start();
      }
      return producer;
    }
  }
}
